package cn.itcast.model.statistics

import cn.itcast.model.base.BaseModel
import cn.itcast.model.statistics.BirthdayModel.spark
import org.apache.spark.sql.{Column, DataFrame}

object CycleModel extends BaseModel{
  override def setAppName(): String = "CycleModel"

  override def importSparkEnv(): Unit = {}
  import spark.implicits._
  import org.apache.spark.sql.functions._
  override def setTagId(): Int = 405

  override def computeTag(fiveRuleDF: DataFrame, hbaseSource: DataFrame): DataFrame = {
    // fiveRuleDF: id: 107 rule: 0-7
    // 将5级规则中的规则信息更换为开始和结束的天数.
    val fiveDF: DataFrame = fiveRuleDF.map(row => {
      val id: String = row.getAs[Int]("id").toString
      val rule: String = row.getAs[String]("rule").toString
      // 将时间段切分为2列
      val arr: Array[String] = rule.split("-")
      val start = arr(0)
      val end = arr(1)
      (id, start, end)
    }).collect().toList.toDF("tagIds", "start", "end")
    //    fiveDF.show()
    //+------+-----+---+
    //|tagIds|start|end|
    //+------+-----+---+
    //|   107|    0|  7|
    //|   108|    8| 14|
    // 一个用户可能有多个订单.
    // 应该获取最近的一笔订单数据.
    val maxDF: DataFrame = hbaseSource.groupBy('memberId)
      .agg(max('finishTime).as("finishTime"))
    //    maxDF.show()
    //+---------+----------+
    //| memberId|finishTime|
    //+---------+----------+
    //| 13822725|1566056954|
    //| 13823083|1566048648|

    // 将订单的完成时间,转换为距今的一个时间(天数)
    val dayNum: Column = datediff(
      //当前时间
      current_timestamp(),
      //订单完成时间
      from_unixtime(maxDF.col("finishTime"))
    )
    val dayDF: DataFrame = maxDF.select('memberid.as("userId"), dayNum.as("dayNum"))
    dayDF.show()
    //+---------+------+
    //|   userId|dayNum|
    //+---------+------+
    //| 13822725|   271|
    //| 13823083|   271|

    // 对数据进行关联,获取标签ID
    //标签最大到4月，271天超出了，关联不上，也就打不上标签
    dayDF.join(fiveDF)
      .where('dayNum.between('start, 'end))
      .select('userId, 'tagIds)
  }
  def main(args: Array[String]): Unit = {
    val newDF: DataFrame = executeModel()
    //saveData(newDF)
  }
}
